package com.binarysoft.service.common.mq;

import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.xml.bind.JAXBElement;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.pool.PooledConnectionFactory;

import common.framework.dsb.AbstractServiceBean;
import common.framework.dsb.annotation.DSBService;
import common.framework.dsb.service.ServiceContext;
import common.framework.jaxb.JAXBTool;
import common.framework.log.Logger;

@DSBService(name = "DSB/MessageQueueService")
public class MessageQueueServiceBean extends AbstractServiceBean implements MessageQueueService {

	/**
	 * key: pool_name, one pool_name means one active-mq instance.
	 * MQService对象封装了单个active-mq 实例的配置参数
	 */
	private Map<String, MQService> mqservices = new Hashtable<String, MQService>();

	/**
	 * key: pool_name, one pool one PooledConnectionFactory.
	 */
	private Map<String, PooledConnectionFactory> factories = new Hashtable<String, PooledConnectionFactory>();

	/**
	 * key: pool_name + queue_name
	 */
	private Map<String, MQDestination> destinations = new Hashtable<String, MQDestination>();

	/**
	 * 一个接收队列，一个队列接收器，队列接收器不负责处理消息，仅把消息放进缓存队列中，消息的处理由多个MessageProcessor并行处理。
	 * key: pool_name + queue_name
	 */
	private Map<String, QueueListener> listeners = new Hashtable<String, QueueListener>();
	/**
	 * 确保每个队列接收器，在消息服务器中断、重启的情况下，能够自动重连
	 */
	private LinkedList<ConnectTask> reconnectTasks = new LinkedList<ConnectTask>();
	private long checkConnectionInterval = 3000;

	private boolean closed = false;

	/**
	 * 消息缓冲区，需要控制内存溢出的情况
	 */
	private LinkedList<MsgWrapper> penddingMsgs = new LinkedList<MsgWrapper>();
	private long maxPenddingMsgs = 1000;

	@Override
	public void start(ServiceContext serviceContext) throws Exception {
		super.start(serviceContext);
		serviceContext.loadConfigProperties("mq-pool.properties");
		checkConnectionInterval = Integer.parseInt(this.serviceContext.getProperty("checkConnectionInterval")) * 1000;
		Logger.log(Logger.LOW_LEVEL, "checkConnectionInterval=" + checkConnectionInterval);
		maxPenddingMsgs = Long.parseLong(this.serviceContext.getProperty("max.pending.length"));
		int msgWorkerCount = Integer.parseInt(this.serviceContext.getProperty("msg.worker.count"));

		loadMQServicesConfig("mq-services.xml");
		initMQPools();

		QueueReconnector reconnector = new QueueReconnector();
		reconnector.setName("MQ-consumer-reconnector");
		reconnector.start();

		for (int i = 0; i < msgWorkerCount; i++) {
			MessageProcessor processor = new MessageProcessor();
			processor.setName("Queue.msg.processor-" + i);
			processor.start();
		}
	}

	@Override
	public void refresh() throws Exception {
	}

	@Override
	public void close() throws Exception {
		closed = true;
	}

	@Override
	public Map<String, MQService> listPools() throws Exception {
		return new HashMap<String, MQService>(mqservices);
	}

	@Override
	public Map<String, MQDestination> listDestinations() throws Exception {
		return new HashMap<String, MQDestination>(destinations);
	}

	@Override
	public Collection<ConnectTask> listReconnectTask() throws Exception {
		return new ArrayList<ConnectTask>(reconnectTasks);
	}

	@Override
	public int pendingMsgs() {
		return penddingMsgs.size();
	}

	@Override
	public void receive(String pool_name, String queue, MessageListener messageListener) throws Exception {
		if (!mqservices.containsKey(pool_name)) {
			throw new Exception("Pool [" + pool_name + "] not found in mq-service.xml");
		}
		String key = pool_name + queue;
		MQDestination dest = destinations.get(key);
		if (dest == null) {
			throw new Exception("Queue [" + queue + "] not found in mq-service.xml");
		}
		QueueListener queueListener = listeners.get(key);
		if (queueListener == null) {
			// create MQ listener
			Logger.log(Logger.FUNCTION_LEVEL, "create MQ listener:" + pool_name + queue);
			queueListener = new QueueListener();
			queueListener.poolName = pool_name;
			queueListener.queueName = queue;
			listeners.put(key, queueListener);
		}
		// 如果重复调用，则覆盖先前的messageListener
		queueListener.delegate = messageListener;

		// Asynchronous start connection
		queueListener.startConnection();
	}

	@Override
	public void stop(String pool_name, String queue) throws Exception {
		String key = pool_name + queue;
		Logger.log(Logger.FUNCTION_LEVEL, "client stop: " + key);
		if (!mqservices.containsKey(pool_name)) {
			throw new Exception("Pool [" + pool_name + "] not found in mq-service.xml");
		}
		MQDestination dest = destinations.get(key);
		if (dest == null) {
			throw new Exception("Queue [" + queue + "] not found in mq-service.xml");
		}
		QueueListener queueListener = listeners.get(key);
		if (queueListener == null) {
			throw new Exception("QueueListener [" + key + "] not found");
		}
		queueListener.stopConnection();
	}

	@Override
	public void send(String pool_name, String queue, Map<String, Object> mapMsg) throws Exception {
		send2mq(pool_name, queue, mapMsg);
	}

	@Override
	public void send(String pool_name, String queue, String text) throws Exception {
		send2mq(pool_name, queue, text);
	}

	private void send2mq(String pool_name, String queue, Object message) throws Exception {
		Logger.log(Logger.LOW_LEVEL, "pool:" + pool_name + " queue:" + queue + " msg:" + message.toString());
		PooledConnectionFactory connectionFactory = factories.get(pool_name);
		if (connectionFactory == null) {
			throw new Exception("Pool not found:" + pool_name);
		}
		MQDestination mqDestination = destinations.get(pool_name + queue);
		if (mqDestination == null) {
			throw new Exception("Queue not found:" + queue);
		}

		Connection connection = null;
		try {
			// 从连接池工厂中获取一个连接
			connection = connectionFactory.createConnection();
			// false 参数表示 为非事务型消息，后面的参数表示消息的确认类型
			Session session = connection.createSession(mqDestination.isTransaction, mqDestination.sessionAckMode);
			Destination destination = null;
			if (mqDestination.isTopic) {
				destination = session.createTopic(queue);
			} else {
				destination = session.createQueue(queue);
			}
			MessageProducer producer = session.createProducer(destination);
			producer.setDeliveryMode(mqDestination.deliveryMode);

			if (message instanceof Map) {
				Map map = (Map) message;
				MapMessage mapMessage = session.createMapMessage();
				if (map != null && !map.isEmpty()) {
					Set<String> keys = map.keySet();
					for (String key : keys) {
						mapMessage.setObject(key, map.get(key));
					}
				}
				producer.send(mapMessage);
			} else {
				TextMessage textMessage = session.createTextMessage(message.toString());
				producer.send(textMessage);
			}
			// producer.close();
			session.close();
		} finally {
			if (connection != null) {
				connection.close();
			}
		}
	}

	private void loadMQServicesConfig(String configFileName) throws Exception {
		URL configFileURL = serviceContext.getConfigFile(configFileName);
		String packageName = this.getClass().getPackage().getName();

		JAXBElement<MQServiceList> jaxbElement = (JAXBElement<MQServiceList>) JAXBTool.unmarshal(configFileURL, packageName, this.getClass().getClassLoader());
		MQServiceList mqList = jaxbElement.getValue();

		for (MQService mq : mqList.mqService) {
			String poolName = mq.getPoolName().trim();
			if (mqservices.containsKey(poolName)) {
				throw new Exception("Error duplicated pool name: " + poolName);
			}
			mqservices.put(poolName, mq);
		}
	}

	private void initMQPools() throws Exception {
		for (MQService mq : mqservices.values()) {
			ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(mq.getBrokerUrl());
			PooledConnectionFactory connectionFactory = new PooledConnectionFactory(factory);

			connectionFactory.setBlockIfSessionPoolIsFull(mq.isBlockIfSessionPoolIsFull());
			connectionFactory.setCreateConnectionOnStartup(mq.isCreateConnectionOnStartup());
			connectionFactory.setExpiryTimeout(mq.getExpiryTimeout());
			connectionFactory.setIdleTimeout(mq.getIdleTimeout());
			connectionFactory.setMaxConnections(mq.getMaxConnections());
			connectionFactory.setMaximumActiveSessionPerConnection(mq.getMaxActiveSessionPerConnection());
			connectionFactory.setTimeBetweenExpirationCheckMillis(mq.getTimeBetweenExpirationCheckMillis());

			factories.put(mq.getPoolName(), connectionFactory);
			connectionFactory.start();

			List<MQDestination> dests = mq.getDestination();
			for (MQDestination dest : dests) {
				destinations.put(mq.getPoolName() + dest.queueName, dest);
			}
		}
	}

	class QueueListener implements MessageListener, ExceptionListener {

		private String poolName = null;
		private String queueName = null;
		// message callback
		private MessageListener delegate = null;

		// MQ control variables
		private boolean queueConnected = false; // 默认是没有连接
		private Connection connection = null;
		private Session session = null;
		private MessageConsumer consumer = null;

		@Override
		public void onMessage(Message message) {
			int currentLength = penddingMsgs.size();
			if (currentLength >= maxPenddingMsgs) {
				Logger.log(Logger.FATAL_LEVEL, "Message ignored!!!");
				return;
			}
			// 由于message.setProperty()方法不可用(read only)，这里每次创建MsgWrapper对象，比较耗性能
			MsgWrapper wrapper = new MsgWrapper();
			wrapper.msg = message;
			wrapper.pool_name = poolName;
			wrapper.queue_name = queueName;
			wrapper.delegate = delegate;
			synchronized (penddingMsgs) {
				penddingMsgs.addLast(wrapper);
				penddingMsgs.notifyAll(); // 通知消息处理器
			}
		}

		@Override
		public void onException(JMSException jmsEx) {
			Logger.log(Logger.FATAL_LEVEL, "Reconnect task created due to:" + jmsEx.getMessage());

			// release previous resources.
			stopConnection();
			// try new message connection.
			startConnection();
		}

		private void stopConnection() {
			// 接收通道关闭
			queueConnected = false;
			try {
				if (consumer != null)
					consumer.close();
			} catch (JMSException e) {
				Logger.printStackTrace(Logger.FUNCTION_LEVEL, "close consumer", e);
			}
			try {
				if (session != null)
					session.close();
			} catch (JMSException e) {
				Logger.printStackTrace(Logger.FUNCTION_LEVEL, "close session", e);
			}
			try {
				if (connection != null)
					connection.close();
			} catch (JMSException e) {
				Logger.printStackTrace(Logger.FUNCTION_LEVEL, "close connection", e);
			}
		}

		private void startConnection() {

			if (queueConnected) {
				// 如果已经处于连接状态，无需重新连接
				return;
			}

			ConnectTask task = new ConnectTask();
			task.setPoolName(poolName);
			task.setQueueName(queueName);
			task.setKey(poolName + queueName);

			synchronized (reconnectTasks) {
				reconnectTasks.addLast(task);
			}// synchronized
		}
	}

	private class MessageProcessor extends Thread {
		public void run() {
			while (!closed) {
				MsgWrapper wrapper = null;
				if (penddingMsgs.size() > 0) {
					synchronized (penddingMsgs) {
						// double check
						if (penddingMsgs.size() > 0) {
							wrapper = penddingMsgs.removeFirst();
						}
					}// synchronized
				}
				if (wrapper != null) {
					try {
						Message msg = wrapper.msg;
						wrapper.delegate.onMessage(msg);
					} catch (Throwable e) {
						Logger.printStackTrace(Logger.FATAL_LEVEL, "MessageProcessor", e);
					}
				} else {
					synchronized (penddingMsgs) {
						try {
							penddingMsgs.wait();
						} catch (InterruptedException e) {
						}
					}// synchronized
				}

			}// while
		}
	} // 消息处理器

	/**
	 * Queue re-connector. 队列连接接收出现异常后，需要重新从连接池中获取连接，并且重新连接消息服务器.
	 * 
	 * @author david
	 * 
	 */
	private class QueueReconnector extends Thread {

		public void run() {
			while (!closed) {
				ConnectTask task = null;
				if (reconnectTasks.size() > 0) {
					synchronized (reconnectTasks) {
						// double check
						if (reconnectTasks.size() > 0) {
							task = reconnectTasks.removeFirst();
						}
					}// synchronized
				}
				if (task != null) {
					try {
						reconnect(task);
					} catch (Throwable e) {
						Logger.printStackTrace(Logger.FATAL_LEVEL, "Consumer reconnect", e);
						// 重连失败，把任务重新放入队列， 等待下一个检测周期
						synchronized (reconnectTasks) {
							reconnectTasks.addLast(task);
						}// synchronized
					}
				}
				try {
					sleep(checkConnectionInterval);
				} catch (InterruptedException e) {
				}
			}// while
		} // run
	}// ConsumerReconnector

	private void reconnect(ConnectTask task) throws Exception {

		String key = task.getPoolName() + task.getQueueName();
		MQDestination dest = destinations.get(key);
		if (dest == null) {
			// 消息队列没有配置
			Logger.log(Logger.FUNCTION_LEVEL, "[" + key + "] not found!");
			return;
		}
		QueueListener listener = listeners.get(key);
		if (listener == null) {
			// 没有消息监听器，则无需建立消息队列连接
			return;
		}

		PooledConnectionFactory connectionFactory = factories.get(task.getPoolName());
		// 从连接池工厂中获取一个连接
		Connection connection = connectionFactory.createConnection();
		connection.setExceptionListener(listener);
		connection.start();
		// false 参数表示 为非事务型消息，后面的参数表示消息的确认类型
		Session session = connection.createSession(dest.isTransaction, dest.getSessionAckMode());
		Destination destination = null;
		if (dest.isTopic) {
			destination = session.createTopic(dest.queueName);
		} else {
			destination = session.createQueue(dest.queueName);
		}
		MessageConsumer consumer = session.createConsumer(destination);
		consumer.setMessageListener(listener);

		// control variables
		listener.connection = connection;
		listener.session = session;
		listener.consumer = consumer;
		listener.queueConnected = true; // 置为连接状态

	}

}
